package org.lql.chaos.kafka.example.controller;

import org.lql.chaos.kafka.example.enums.PartitionEnum;
import org.lql.chaos.kafka.example.pojo.Order;
import org.lql.chaos.kafka.example.producer.KafkaProducerExampleService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

/**
 * @author: liangqinglong
 * @date: 2025-08-10 01:20
 * @description:
 **/
@RestController
@RequestMapping("/kafka-example")
public class KafkaExampleController {

	@Autowired
	private KafkaProducerExampleService kafkaProducerExampleService;

	@RequestMapping("/send-simple-message-by-string-serializer")
	public String sendSimpleMessageByStringSerializer() {
		kafkaProducerExampleService.sendSimpleMessageByStringSerializer(
				"topic-string", "string", "hello kafka simple-message-by-string-serializer", PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-complex-object-message-by-string-serializer")
	public String sendComplexObjectMessageByStringSerializer() {
		Order order = Order.mock();
		kafkaProducerExampleService.sendComplexObjectMessageByStringSerializer(
				"topic-string", order.getOrderId(), order, PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-simple-message-by-bytes-serializer")
	public String sendSimpleMessageByBytesSerializer() {
		kafkaProducerExampleService.sendSimpleMessageByBytesSerializer(
				"topic-bytes", "bytes", "hello world simple-message-by-bytes-serializer".getBytes(), PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-complex-object-message-by-bytes-serializer")
	public String sendComplexObjectMessageByBytesSerializer() {
		Order order = Order.mock();
		kafkaProducerExampleService.sendComplexObjectMessageByBytesSerializer(
				"topic-bytes", order.getOrderId(), order, PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-simple-message-by-json-serializer")
	public String sendSimpleMessageByJsonSerializer() {
		kafkaProducerExampleService.sendSimpleMessageByJsonSerializer(
				"topic-json", "json", "hello kafka send-simple-message-by-json-serializer", PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-complex-object-message-by-json-serializer")
	public String sendComplexObjectMessageByJsonSerializer() {
		Order order = Order.mock();
		kafkaProducerExampleService.sendComplexObjectMessageByJsonSerializer(
				"topic-json", order.getOrderId(), order, PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-simple-message-by-custom-serializer")
	public String sendSimpleMessageByCustomSerializer() {
		kafkaProducerExampleService.sendSimpleMessageByCustomSerializer(
				"topic-kryo", "custom", "hello kafka send-simple-message-by-custom-serializer", PartitionEnum.HASH.getName());
		return "success";
	}

	@RequestMapping("/send-complex-object-message-by-custom-serializer")
	public String sendComplexObjectMessageByCustomSerializer() {
		Order order = Order.mock();
		kafkaProducerExampleService.sendComplexObjectMessageByCustomSerializer(
				"topic-kryo", order.getOrderId(), order, PartitionEnum.HASH.getName());
		return "success";
	}
}
